package com.example.mq2.mqclient;

import com.example.mq2.common.BinaryTool;
import com.example.mq2.common.MqException;
import com.example.mq2.network.BasicResponse;
import com.example.mq2.network.MessageCallbackArguments;
import com.example.mq2.network.Request;
import com.example.mq2.network.Response;
import org.springframework.web.context.request.RequestScope;

import java.io.*;
import java.net.Socket;
import java.net.SocketException;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class Connection {

    private Socket clientSocket;
    //一个 connection 管理多个 channel，key 是 channelId
    private ConcurrentHashMap<String, Channel> channelMap = new ConcurrentHashMap<>();
    //统一管理输入输出
    private InputStream inputStream;
    private OutputStream outputStream;
    private DataInputStream dataInputStream;
    private DataOutputStream dataOutputStream;
    //使用一个扫描线程，不断的获取服务器传来的响应
    private Thread readRespThread;
    private ExecutorService callbackHandlerPool = Executors.newFixedThreadPool(4);

    public Connection(String host, int port) throws IOException {
        this.clientSocket = new Socket(host, port);
        this.inputStream = clientSocket.getInputStream();
        this.outputStream = clientSocket.getOutputStream();
        this.dataInputStream = new DataInputStream(inputStream);
        this.dataOutputStream = new DataOutputStream(outputStream);

        //这个扫描线程要做的就是不断扫描，看服务器那边是否传来响应
        readRespThread = new Thread(() -> {
            while(!clientSocket.isClosed()) {
                try {
                    //1.读取服务器响应
                    Response response = readResponse();
                    //2.根据响应的不同类型，将响应发送给 channel，解除 channel 的阻塞等待响应
                    handlerChannelResponse(response);
                } catch (SocketException e) {
                    System.out.println("[Connection] 客户端连接断开");
                }
                catch (IOException | MqException e) {
                    throw new RuntimeException(e);
                }
            }
        });

        readRespThread.setDaemon(true);
        readRespThread.start();
    }

    public void close() throws IOException {
       clientSocket.close();
       channelMap.clear();
       callbackHandlerPool.shutdown();
    }


    /**
     * 读取响应
     * @return
     * @throws IOException
     */
    private Response readResponse() throws IOException {
        int type = dataInputStream.readInt();
        int length = dataInputStream.readInt();
        byte[] payload = new byte[length];
        int actualSize = dataInputStream.read(payload);
        if(length != actualSize) {
            System.out.println("[Connection] 读取响应格式非法！expectedSize=" + length + ", actualSize=" + actualSize);
        }

        Response response = new Response();
        response.setType(type);
        response.setLength(length);
        response.setPayload(payload);
        return response;
    }

    /**
     * 写请求给服务器
     * @param request
     * @throws IOException
     */
    public void writeRequest(Request request) throws IOException {
        dataOutputStream.writeInt(request.getType());
        dataOutputStream.writeInt(request.getLength());
        dataOutputStream.write(request.getPayload());
        dataOutputStream.flush();
    }

    /**
     * 区分回调和普通响应
     * @param response
     */
    private void handlerChannelResponse(Response response) throws MqException {
        if(response.getType() == 0xc) {
            MessageCallbackArguments messageCallbackArguments = (MessageCallbackArguments) BinaryTool.fromBytes(response.getPayload());
            //从 channel 中获取回调，在线程池中执行
            Channel channel = channelMap.get(messageCallbackArguments.getChannelId());
            if(channel == null) {
                throw new MqException("[Connection] Channel 不存在(0xc)！");
            }
            //使用线程池来执行回调
            callbackHandlerPool.submit(() -> {
                try {
                    channel.getConsumer().handlerDeliver(messageCallbackArguments.getConsumerTag(), messageCallbackArguments.getBasicProperties(),
                            messageCallbackArguments.getBody());
                } catch (IOException e) {
                    throw new RuntimeException(e);
                }
            });
        } else {
            //这里是普通响应，交给 Channel 去处理(解除 channel 发送请求后的阻塞等待)
            BasicResponse basicResponse = (BasicResponse) BinaryTool.fromBytes(response.getPayload());
            //获取 channel
            Channel channel = channelMap.get(basicResponse.getChannelId());
            if(channel == null) {
                throw new MqException("[Connection] Channel 不存在！");
            }
            channel.putResponse(basicResponse);
        }
    }

    /**
     * 创建 Channel
     * @return
     */
    public Channel createChannel() throws IOException {
        //1.创建channel
        String channelId = "C-" + UUID.randomUUID().toString();
        Channel channel = new Channel(channelId, this);
        //2.将 channel 放入 connection 管理的 哈希表中
        channelMap.put(channelId, channel);
        //3.将创建 channel 的消息告知服务器
        channel.create(channel);
        return channel;
    }


}
